package com.atguigu.day03;

import com.atguigu.day02.Example2;
import org.apache.commons.collections.map.HashedMap;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.util.List;

/**
 *  尝试使用keyedProcessFunction实现滚动
 */
public class Example5_1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Example2.Event> dataStreamSource = env.addSource(new Example2.MySource());
        dataStreamSource.keyBy(x->x.name).process(new KeyedProcessFunction<String, Example2.Event, List<Example2.Event>>() {
            private HashedMap map = new HashedMap();
            @Override
            public void processElement(Example2.Event value, Context ctx, Collector<List<Example2.Event>> out) throws Exception {
                ctx.timerService().registerProcessingTimeTimer(ctx.timerService().currentProcessingTime()+10*1000);
            }

        }).print();

        env.execute();
    }
}
